package com.zjh.rocketmq.producer;

import cn.hutool.core.util.IdUtil;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.io.Serializable;

@Component
public class RocketMqProducer {
    private final Logger logger = LoggerFactory.getLogger(RocketMqProducer.class);


    @Autowired
    private RocketMQTemplate rocketMQTemplate;


    /**
     * MQ半消息
     *
     * @param topic target topic
     * @param tag   topic's tag
     * @param msg   message
     * @return send status
     */
    public <T extends Serializable> SendStatus txSend(T msg, String topic, String tag) {
        String destination = String.format("%s:%s", topic, tag);

        Message<T> message = MessageBuilder.withPayload(msg)
                .setHeader("KEYS", IdUtil.simpleUUID())
                .setHeader("DESTINATION", destination)
                .build();

        TransactionSendResult result =
                rocketMQTemplate.sendMessageInTransaction(destination, message, msg);

        //发送状态
        String sendStatus = result.getSendStatus().name();
        // 本地事务执行状态
        String localTxState = result.getLocalTransactionState().name();
        logger.info("send tx message sendStatus:{},localTXState:{}", sendStatus, localTxState);
        return result.getSendStatus();
    }

    /**
     * 同步消息
     *
     * @param msg
     * @param topic
     * @param tag
     * @param <T>
     * @return
     */
    public <T extends Serializable> SendStatus syncSend(T msg, String topic, String tag) {
        String destination = String.format("%s:%s", topic, tag);

        Message<T> message = MessageBuilder.withPayload(msg)
                .setHeader("KEYS", IdUtil.simpleUUID())
                .setHeader("DESTINATION", destination)
                .build();
        return rocketMQTemplate.syncSend(destination, message).getSendStatus();
    }

    /**
     * 异步消息
     *
     * @param msg
     * @param topic
     * @param tag
     * @param <T>
     * @return
     */
    public <T extends Serializable> void asyncSend(T msg, String topic, String tag) {
        String destination = String.format("%s:%s", topic, tag);

        Message<T> message = MessageBuilder.withPayload(msg)
                .setHeader("KEYS", IdUtil.simpleUUID())
                .setHeader("DESTINATION", destination)
                .build();
        rocketMQTemplate.asyncSend(destination, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("============================发送成功=================================");
            }

            @Override
            public void onException(Throwable e) {
                System.out.println("============================发送失败================================="+e.getMessage());
            }
        });
    }



}
